package com.bottomlord.reactice.streams;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * @author ChenYue
 * @date 2021/3/31 22:15
 */
public class BusinessSubscriber<T> implements Subscriber<T> {
    private Subscription subscription;
    private int count = 0;
    private final long maxRequest;

    public BusinessSubscriber(long maxRequest) {
        this.maxRequest = maxRequest;
    }

    @Override
    public void onSubscribe(Subscription subscription) {
        this.subscription = subscription;
        this.subscription.request(maxRequest);
    }

    @Override
    public void onNext(T t) {
        if (count++ > 2) {
            subscription.cancel();;
            return;
        }

        System.out.println("收到数据：" + count);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("遇到异常：" + throwable);
    }

    @Override
    public void onComplete() {
        System.out.println("接收完成。");
    }
}
